package com.xiguaji.tiktok.task.fansportrait.service;

import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerLogRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerTypeRepository;
import com.gugee.tiktok.common.model.dto.SlimBloggerDto;
import com.gugee.tiktok.common.model.mysql.Aweme;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.data.common.msg.FansPortraitMessage;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-10-28 15:16
 */
@Service
@Slf4j
public class FansPortraitService extends AbstractProducerHandler<FansPortraitMessage> {


    @Autowired
    BloggerLogRepository bloggerLogRepository;

    @Autowired
    @Qualifier("kafkaTemplateFansPortrait")
    private KafkaTemplate<Long, FansPortraitMessage> kafkaTemplateFansPortrait;
    @Autowired
    BloggerMainRepository bloggerMainRepository;
    @Autowired
    AppConfig appConfig;
    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    BloggerTypeRepository bloggerTypeRepository;
    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");

    /**
     * 这里不做处理
     * @param bloggerLog
     * @param taskParam
     * @return
     */
    @Override
    public List<FansPortraitMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam){
        return null;
    }
    /**
     * 查询具体页内的记录，即需要发送到kafka队列的博主id和视频id集合
     * @param bloggerLog 业务类型数据的采集页
     * @param taskParam  任务参数，包含了每页的页码大小
     * @return
     */
    public List<SlimBloggerDto> findBloggerObject(BloggerLog bloggerLog, TaskParam taskParam) {

        //查询本次需要进行更新的博主
        List<SlimBloggerDto> bloggers = bloggerMainRepository.findSlimBloggersFollowerAndCount(appConfig.getSysCollectFollowerCount(),
                ((int)bloggerLog.getCurrentBloggerId() - 1) * taskParam.getRecords(),taskParam.getRecords());
        return bloggers;
    }

    /**
     *
     * @param messages
     * @param topic
     */
    @Override
    public void sendMsg(List<FansPortraitMessage> messages,String topic) {
        messages.forEach( msg ->{
            msg.setRetryTimes(appConfig.getRetryFansPortraitTimes());
            msg.setRetryLimitTimes(appConfig.getRetryFansPortraitTimes());
            kafkaTemplateFansPortrait.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }

    /**
     * 根据 blogger 查询 粉丝画像分析消息
     * @param bloggers
     */
    public List<FansPortraitMessage> findMessageObject(List<SlimBloggerDto> bloggers) {
        //时间区间范围的视频
        LocalDateTime localDateTime =  LocalDateTime.now().minusDays(appConfig.getLatestDays() + 2);
        int condition = Integer.parseInt(localDateTime.format(formatter));
        //视频粉丝画像最近更新时间
        long lastUpdateTime = (System.currentTimeMillis() / 1000 - appConfig.getFansPortraitIntervalHours() * 3600);
        List<FansPortraitMessage> fansPortraitMessages;
        if(bloggers != null && bloggers.size() > 0){
            fansPortraitMessages = new ArrayList<>(bloggers.size() * 50);
            //查询博主下需要被更新的视频
            List<Aweme> awemes;
            List<Long> bloggerIds = new ArrayList<>(12);
            //批量查
            for(int i = 0; i < bloggers.size(); ++i){
                if(i == bloggers.size() - 1 || bloggerIds.size() == 10){
                    if(bloggerIds.size() == 0){
                        bloggerIds.add(bloggers.get(i).getUid());
                    }
                    awemes =  awemeRepository.getFansPortraitAwemeByRecentTimes(bloggerIds,condition,lastUpdateTime,appConfig.getFansPortraitDisabilityRate());
                    awemes.forEach(aweme -> {
                        FansPortraitMessage fansPortraitMessage = new FansPortraitMessage();
                        fansPortraitMessage.setUid(aweme.getOUid());
                        fansPortraitMessage.setId(aweme.getId());
                        fansPortraitMessage.setAwemeId(aweme.getOAwemeId());
                        fansPortraitMessages.add(fansPortraitMessage);
                    });
                    bloggerIds = new ArrayList<>(12);
                }else{
                    bloggerIds.add(bloggers.get(i).getUid());
                }
            }
        }else{
            fansPortraitMessages = new ArrayList<>();
        }
        return fansPortraitMessages;
    }

    /**
     * 发送
     * @param bloggerLogType
     * @param who
     * @param taskParam
     */
    @Override
    public void sendMsg2Kafka(int bloggerLogType, String who,TaskParam taskParam){
        long start=System.currentTimeMillis();
        //获取将发送的位置
        BloggerLog bloggerLog = getBloggerLog(bloggerLogType,who);
        if(taskParam.getStartPage() == bloggerLog.getCurrentBloggerId()){
            String timeStr2 = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            bloggerLog.setFirstPageStartTime(timeStr2);
        }
        //获取应该发送的消息
        List<SlimBloggerDto> bloggers = findBloggerObject(bloggerLog,taskParam);
        //记录下一次发送的位置
        long nextPage = bloggerLog.getCurrentBloggerId() % taskParam.getPagesLimit() + 1;
        if (CollectionUtils.isEmpty(bloggers) || nextPage == taskParam.getStartPage()) {
            nextPage = taskParam.getStartPage();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            try {
                long mills=System.currentTimeMillis()-sdf.parse(bloggerLog.getFirstPageStartTime()).getTime();
                bloggerLog.setMoreInfo( "起始页="+taskParam.getStartPage()+",需轮询总页数="+bloggerLog.getCurrentBloggerId()+",上一轮花费(分钟)="+mills/1000/60+",上一轮开始时间="+bloggerLog.getFirstPageStartTime());
                log.warn("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
                XxlJobLogger.log("页码={} ，无数据,下一次将从{}页开始",bloggerLog.getCurrentBloggerId(),nextPage);
            } catch (ParseException e) {
                e.printStackTrace();
            }
        }

        List<FansPortraitMessage> messages = findMessageObject(bloggers);
        sendMsg(messages,taskParam.getTopic());
        //发送消息后，设置下一次发送位置
        bloggerLog.setCurrentBloggerId(nextPage);
        bloggerLogRepository.save(bloggerLog);
        log.info(" who={},fansMsgSize={} , bloggerSize={},page={}, pageSize={}, record from {} to {},times={}秒",
                who,messages.size(),bloggers.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),
                (bloggerLog.getCurrentBloggerId() - 1) * taskParam.getRecords(),
                bloggerLog.getCurrentBloggerId() * taskParam.getRecords(),
                (System.currentTimeMillis() - start) / 1000);
        XxlJobLogger.log(" fansMsgSize={} , bloggerSize={},page={}, pageSize={},  record from  {} to {},times={}秒",
                messages.size(),bloggers.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),
                (bloggerLog.getCurrentBloggerId() - 1) * taskParam.getRecords(),
                bloggerLog.getCurrentBloggerId() * taskParam.getRecords(),
                (System.currentTimeMillis() - start) / 1000);

    }
}
